package org.example.netty.httpserver;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoop;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpResponse;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class FlushOptimizedOrderHandler extends SimpleChannelInboundHandler<FullHttpResponse> {

    private static final ConcurrentMap<EventLoop, Flusher> flusherLookup = new ConcurrentHashMap<>();

    private static class FlushItem {
        final ChannelHandlerContext ctx;
        final Object response;

        private FlushItem(ChannelHandlerContext ctx, Object response) {
            this.ctx = ctx;
            this.response = response;
        }
    }

    private final class Flusher implements Runnable {
        final EventLoop eventLoop;
        final ConcurrentLinkedQueue<FlushItem> queued = new ConcurrentLinkedQueue<>();
        final AtomicBoolean running = new AtomicBoolean(false);
        final HashSet<ChannelHandlerContext> channels = new HashSet<>();
        final List<FlushItem> flushed = new ArrayList<>();
        int runsSinceFlush = 0;
        int runsWithNoWork = 0;

        private Flusher(EventLoop eventLoop) {
            this.eventLoop = eventLoop;
        }

        void start() {
            if (!running.get() && running.compareAndSet(false, true)) {
                this.eventLoop.execute(this);
            }
        }

        @Override
        public void run() {

            boolean doneWork = false;
            FlushItem flush;
            while (null != (flush = queued.poll())) {
                channels.add(flush.ctx);
                // 因为是异步的，主线程已经释放过1次，如果在释放1次，就会报错
                // write的时候，会去分配ByteBuf，然后会去进行内存泄漏检测
                flush.ctx.write(flush.response, flush.ctx.voidPromise());
                flushed.add(flush);
                doneWork = true;
            }

            runsSinceFlush++;

            if (!doneWork || runsSinceFlush > 2 || flushed.size() > 50) {
                for (ChannelHandlerContext channel : channels) {
                    channel.flush();
                }

                channels.clear();
                flushed.clear();
                runsSinceFlush = 0;
            }

            if (doneWork) {
                runsWithNoWork = 0;
            } else {
                // either reschedule or cancel
                if (++runsWithNoWork > 5) {
                    running.set(false);
                    if (queued.isEmpty() || !running.compareAndSet(false, true)) {
                        return;
                    }
                }
            }

            eventLoop.schedule(this, 10000, TimeUnit.NANOSECONDS);
        }
    }

    private void flush(FlushItem item) {
        EventLoop loop = item.ctx.channel().eventLoop();
        Flusher flusher = flusherLookup.get(loop);
        if (flusher == null) {
            Flusher alt = flusherLookup.putIfAbsent(loop, flusher = new Flusher(loop));
            if (alt != null) {
                flusher = alt;
            }
        }

        flusher.queued.add(item);
        flusher.start();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse msg) throws Exception {
        // 处理消息并发送
        // 增加一次引用
        msg.retain();
        flush(new FlushItem(ctx, msg));
    }
}
